#include "tcp_client_concrete.h"

static void split(std::string src, std::string delim, std::vector<std::string> * outlist)
{
	if (src == "") {
		return;
	}
	int last = 0, index;
	std::string tt;
	index = src.find_first_of(delim, last);
	while (index != std::string::npos) {
		tt = src.substr(last, index - last);
		outlist->push_back(tt);
		last = index + delim.size();
		index = src.find_first_of(delim, last);
	}
	if (index - last > 0) {
		tt = src.substr(last, index - last);
		if (tt != "") {
			outlist->push_back(tt);
		}
	}
}

cmtd::TcpClient::TcpClient()
	: m_bQuit(false), m_hExitEvent(NULL), m_hHeartThread(NULL), m_hWorkerThread(NULL), 
	m_iocp(NULL), m_pool(NULL)  
{
	WSAData wsadata;
	WSAStartup(MAKEWORD(2, 2), &wsadata);
	InitializeCriticalSection(&m_csForClientInstList);
	InitializeConditionVariable(&m_cvForListNotEmpty);
	clear_client_insts();	
	m_bQuit = false;
	m_nSerialNo = SERIALNO_START;
	m_sendDataBuffer = new unsigned char[BUFFER_MAX_SIZE];
	memset(m_sendDataBuffer, 0, BUFFER_MAX_SIZE);
}

int cmtd::TcpClient::Init()
{
	unsigned int uiThreadId = 0;
	if (m_iocp == NULL) {
		m_iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 0);
	}
	if (m_iocp) {		
		if (m_hWorkerThread == NULL) {
			m_hWorkerThread = (HANDLE)_beginthreadex(NULL, 0, worker_thread, this, 0, &uiThreadId);
		}
	}
	if (m_hExitEvent == NULL) {
		m_hExitEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
	}
	if (m_hHeartThread == NULL) {
		m_hHeartThread = (HANDLE)_beginthreadex(NULL, 0, keep_alive_thread, this, 0, &uiThreadId);
	}
	if (m_pool == NULL) {
		m_pool = threadpool_init(1, 20);
	}
	return 0;
}

void cmtd::TcpClient::UnInit()
{
	m_bQuit = true;
	if (m_hExitEvent) {
		SetEvent(m_hExitEvent);
		RELEASE_HANDLE(m_hExitEvent);
	}
	if (m_pool) {
		threadpool_destroy(m_pool);
		m_pool = NULL;
	}
	WakeAllConditionVariable(&m_cvForListNotEmpty);
	if (m_hHeartThread) {
		WaitForSingleObject(m_hHeartThread, INFINITE);
		RELEASE_HANDLE(m_hHeartThread);
	}
	PostQueuedCompletionStatus(m_iocp, 0, (ULONG_PTR)NULL, NULL);
	if (m_hWorkerThread) {
		WaitForSingleObject(m_hWorkerThread, INFINITE);
		RELEASE_HANDLE(m_hWorkerThread);
	}
}

cmtd::TcpClient::~TcpClient()
{
	UnInit();
	clear_client_insts();
	DeleteCriticalSection(&m_csForClientInstList);
	RELEASE_HANDLE(m_iocp);
	RELEASE_ARRAY(m_sendDataBuffer);
	WSACleanup();
}

int cmtd::TcpClient::RegisterModule(cmtd::SERVER_INFO server_info_, cmtd::MODULE_INFO_PTR module_info_ptr_, 
	cmtd::fMessageCallback msgCb_, void * userData_)
{
	int result = -1;
	SOCKET sock = INVALID_SOCKET;
	//sock = socket(AF_INET, SOCK_STREAM, 0);
	sock = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
	if (sock == INVALID_SOCKET) {
		return -1;
	}
	struct sockaddr_in server_addr;
	server_addr.sin_family = AF_INET;
	server_addr.sin_port = htons((USHORT)server_info_.uiServerPort);
	inet_pton(AF_INET, server_info_.szServerIp, &server_addr.sin_addr);
	if (connect(sock, (const sockaddr *)&server_addr, sizeof(sockaddr)) != 0) {
		int errCode = WSAGetLastError();
		RELEASE_SOCKET(sock);
		return -1;
	}
	int nHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
	int nInfoLen = sizeof(cmtd::MODULE_INFO);
	cmtd::MEMOBUS_HEAD memoHead;
	MAKE_HEAD(memoHead);
	memoHead.uiCmdType = cmtd::CMD_REGISTER_REQ;
	memoHead.ulSerialNo = m_nSerialNo++;
	memoHead.ulTimeStamp = static_cast<unsigned long>(time(NULL));
	memoHead.ulCmdLen = nInfoLen;
	memoHead.uiVersion = INTERNAL_VERSION;
	int offset = 0;
	memcpy_s(m_sendDataBuffer, BUFFER_MAX_SIZE, &memoHead, nHeadLen);
	offset += nHeadLen;
	memcpy_s(m_sendDataBuffer + nHeadLen, BUFFER_MAX_SIZE - nHeadLen, module_info_ptr_, nInfoLen);
	offset += nInfoLen;
	if (send(sock, reinterpret_cast<const char *>(m_sendDataBuffer), offset, 0) != SOCKET_ERROR) {
		int nRespInfoLen = sizeof(cmtd::REGISTER_RESPONSE_PARAMETER);
		fd_set fd_read;
		struct timeval timeout;
		timeout.tv_sec = 2;
		timeout.tv_usec = 0;
		unsigned char * szRecvBuf = new unsigned char[BUFFER_MAX_SIZE];
		memset(szRecvBuf, 0, BUFFER_MAX_SIZE);
		int nTimeoutCount = 0;
		while (1) {
			FD_ZERO(&fd_read);
			FD_SET(sock, &fd_read);
			int ret = select(sock + 1, &fd_read, NULL, NULL, &timeout);
			if (ret == SOCKET_ERROR) {
				RELEASE_ARRAY(szRecvBuf);
				RELEASE_SOCKET(sock);
				return -1;
			}
			else if (ret == 0) {
				nTimeoutCount++;
				Sleep(10);
				continue;
			}
			else if (FD_ISSET(sock, &fd_read)) {
				int nRecvLen = recv(sock, reinterpret_cast<char *>(szRecvBuf), BUFFER_MAX_SIZE, 0);
				if (nRecvLen == SOCKET_ERROR) {
					RELEASE_ARRAY(szRecvBuf);
					RELEASE_SOCKET(sock);
					return -1;
				}
				else if (nRecvLen == 0) {
					continue;
				}
				else if (nRecvLen > 0) {
					if (nRecvLen == nRespInfoLen + nHeadLen) {
						cmtd::MEMOBUS_HEAD recvMemoHead;
						memcpy_s(&recvMemoHead, nHeadLen, szRecvBuf, nHeadLen);
						if (check_valid_head(recvMemoHead)) {
							cmtd::REGISTER_RESPONSE_PARAMETER regRespParam;
							memcpy_s(&regRespParam, nRespInfoLen, szRecvBuf + nHeadLen, nRespInfoLen);
							if (regRespParam.nResult == 0) {
								strncpy_s(module_info_ptr_->szSession,
									sizeof(module_info_ptr_->szSession),
									regRespParam.register_parameter.szSession,
									strlen(regRespParam.register_parameter.szSession));
								cmtd::CLIENT_CONTEXT_PTR pClientContext = new cmtd::CLIENT_CONTEXT();
								pClientContext->clt_sock_num = sock;
								pClientContext->clt_status = module_info_ptr_->nStatus;
								pClientContext->clt_type = module_info_ptr_->nType;
								pClientContext->clt_work_type = module_info_ptr_->nDealType; 
								pClientContext->clt_session = regRespParam.register_parameter.szSession;
								pClientContext->clt_register_time = (long)time(NULL);
								pClientContext->clt_last_active_time = pClientContext->clt_register_time;			
								if (associationWithCompletionPort(sock, m_iocp, (DWORD)pClientContext)) {
									DWORD dwRecvBytes;
									DWORD dwFlags = 0;
									IO_CONTEXT_PTR pIoContext = new IO_CONTEXT();
									int err = WSARecv(sock, &pIoContext->wsaBuf, 1, &dwRecvBytes, &dwFlags, 
										&pIoContext->overlapped, NULL);
									if (err == SOCKET_ERROR && WSAGetLastError() != WSAGetLastError()) {
										RELEASE(pIoContext);
										RELEASE(pClientContext);
									}
									else {
										add_client_inst(pClientContext->clt_session, pClientContext);
										pClientContext->clt_io_contexts.push_back(pIoContext);
										pClientContext->msgCB = msgCb_;
										pClientContext->userDataForMsgCB = userData_;

										cmtd::MESSAGE_INFO msgInfo;
										msgInfo.nMsgType = cmtd::CMD_ONLINE;
										strncpy_s(msgInfo.szSession, sizeof(msgInfo.szSession),
											module_info_ptr_->szSession, strlen(module_info_ptr_->szSession));
										cmtd::ONLINE_PARAMETER onlineParam;
										onlineParam.nStatus = module_info_ptr_->nStatus;
										onlineParam.nType = module_info_ptr_->nType;
										onlineParam.nDealType = module_info_ptr_->nDealType;
										strncpy_s(onlineParam.szSession, sizeof(onlineParam.szSession),
											module_info_ptr_->szSession, strlen(module_info_ptr_->szSession));
										msgInfo.msgDataLen = sizeof(onlineParam);
										msgInfo.msgData = new unsigned char[msgInfo.msgDataLen];
										memcpy_s(msgInfo.msgData, msgInfo.msgDataLen, &onlineParam, 
											msgInfo.msgDataLen);
										if (pClientContext->msgCB) {
											pClientContext->msgCB(msgInfo);
										}
										msgInfo.userData = pClientContext->userDataForMsgCB;
										RELEASE_ARRAY(msgInfo.msgData);
										result = 0;
									}
									break;
								}
								else { //association failed
									RELEASE(pClientContext);
								}
							}
							else { //return error from server

							}
						}
					}
				}
			}
			if (nTimeoutCount > 10) {
				break;
			}
		}
		RELEASE_ARRAY(szRecvBuf);
	}
	return result;	
}

int cmtd::TcpClient::UnRegisterModule(cmtd::UNREGISTER_PARAMETER unRegParam_)
{
	int result = -1;
	std::string str_session = unRegParam_.szSession;
	EnterCriticalSection(&m_csForClientInstList);
	do {
		if (m_clientInstList.empty()) {
			break;
		}
		std::map<std::string, cmtd::CLIENT_CONTEXT_PTR>::iterator iter;
		iter = m_clientInstList.find(str_session);
		if (iter != m_clientInstList.end()) {
			CLIENT_CONTEXT_PTR pContext = iter->second;
			if (pContext) {
				SOCKET sock_num = pContext->clt_sock_num;
				if (sock_num != INVALID_SOCKET) {
					cmtd::MEMOBUS_HEAD reqHead;
					unsigned long ulHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
					unsigned long ulParamLen = sizeof(cmtd::UNREGISTER_PARAMETER);
					MAKE_HEAD(reqHead);
					reqHead.uiCmdType = cmtd::CMD_UNREGISTER_REQ;
					reqHead.uiVersion = INTERNAL_VERSION;
					reqHead.ulSerialNo = m_nSerialNo++;
					reqHead.ulCmdLen = ulParamLen;
					reqHead.ulTimeStamp = static_cast<unsigned long>(time(NULL));
					unsigned long ulBufLen = ulHeadLen + ulParamLen + 1;
					unsigned char * pSendBuf = new unsigned char [ulBufLen];
					memset(pSendBuf, 0, ulBufLen);
					int offset = 0;
					memcpy_s(pSendBuf, ulHeadLen, &reqHead, ulHeadLen);
					offset += ulHeadLen;
					memcpy_s(pSendBuf + offset, ulParamLen, &unRegParam_, ulParamLen);
					offset += ulParamLen;
					if (send(sock_num, (const char *)pSendBuf, offset, 0) != SOCKET_ERROR) {
						RELEASE(pContext);
						m_clientInstList.erase(iter);
						result = 0;
					}
					RELEASE_ARRAY(pSendBuf);
				}
			}
		}
	} while (0);
	LeaveCriticalSection(&m_csForClientInstList);
	return result;
}

int cmtd::TcpClient::FetchTask(cmtd::FETCH_PARAMETER fetchParam_)
{
	int result = -1;
	std::string str_session = fetchParam_.szSession;
	EnterCriticalSection(&m_csForClientInstList);
	do {
		if (m_clientInstList.empty() || str_session == "") {
			break;
		}
		std::map<std::string, cmtd::CLIENT_CONTEXT_PTR>::iterator iter;
		iter = m_clientInstList.find(str_session);
		if (iter != m_clientInstList.end()) {
			cmtd::CLIENT_CONTEXT_PTR pClientContext = iter->second;
			if (pClientContext && pClientContext->clt_sock_num != INVALID_SOCKET) {
				cmtd::MEMOBUS_HEAD reqHead;
				unsigned long ulReqHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
				unsigned long ulFetchParamLen = sizeof(cmtd::FETCH_PARAMETER);
				MAKE_HEAD(reqHead);
				reqHead.uiVersion = INTERNAL_VERSION;
				reqHead.ulSerialNo = m_nSerialNo++;
				reqHead.uiCmdType = cmtd::CMD_FETCH_REQ;
				reqHead.ulCmdLen = sizeof(cmtd::FETCH_PARAMETER);
				reqHead.ulTimeStamp = static_cast<unsigned long>(time(NULL));
				unsigned long ulSendBufLen = ulReqHeadLen + ulFetchParamLen + 1;
				unsigned char * pSendBuf = new unsigned char[ulSendBufLen];
				memset(pSendBuf, 0, ulSendBufLen);
				int offset = 0;
				memcpy_s(pSendBuf, ulReqHeadLen, &reqHead, ulReqHeadLen);
				offset += ulReqHeadLen;
				memcpy_s(pSendBuf + offset, ulFetchParamLen, &fetchParam_, ulFetchParamLen);
				offset += ulFetchParamLen;
				if (send(pClientContext->clt_sock_num, reinterpret_cast<const char *>(pSendBuf),
					offset, 0) != SOCKET_ERROR) {
					result = 0;
				}
				RELEASE_ARRAY(pSendBuf);
			}
		}
	} while (0);
	LeaveCriticalSection(&m_csForClientInstList);
	return result;
}

int cmtd::TcpClient::FinishTask(cmtd::FINISH_PARAMETER finishParam_)
{
	int result = -1;
	std::string str_session = finishParam_.szSession;
	EnterCriticalSection(&m_csForClientInstList);
	do {
		if (str_session == "" || m_clientInstList.empty()) {
			break;
		}
		std::map<std::string, cmtd::CLIENT_CONTEXT_PTR>::iterator iter;
		iter = m_clientInstList.find(str_session);
		if (iter != m_clientInstList.end()) {
			cmtd::CLIENT_CONTEXT_PTR pClientContext = iter->second;
			if (pClientContext && pClientContext->clt_sock_num != INVALID_SOCKET) {
				cmtd::MEMOBUS_HEAD reqHead;
				MAKE_HEAD(reqHead);
				reqHead.uiCmdType = cmtd::CMD_FINISH_REQ;
				reqHead.ulCmdLen = sizeof(cmtd::FINISH_PARAMETER);
				reqHead.ulSerialNo = m_nSerialNo++;
				reqHead.ulTimeStamp = static_cast<unsigned long>(time(NULL));
				unsigned long ulReqHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
				unsigned char * pSendBuf = new unsigned char [BUFFER_MAX_SIZE];
				memset(pSendBuf, 0, BUFFER_MAX_SIZE);
				int offset = 0;
				unsigned long ulFinishParamLen = sizeof(cmtd::FINISH_PARAMETER);
				memcpy_s(pSendBuf, ulReqHeadLen, &reqHead, ulReqHeadLen);
				offset += ulReqHeadLen;
				memcpy_s(pSendBuf + offset, ulFinishParamLen, &finishParam_, ulFinishParamLen);
				offset += ulFinishParamLen;
				if (send(pClientContext->clt_sock_num, reinterpret_cast<const char *>(pSendBuf),
					offset, 0) != SOCKET_ERROR) {
					result = 0;
				}
				RELEASE_ARRAY(pSendBuf);
			}
		}
	} while (0);
	LeaveCriticalSection(&m_csForClientInstList);
	return result;
}

int cmtd::TcpClient::PostTask(cmtd::POST_PARAMETER postParam_)
{
	int result = -1;
	std::string str_session = postParam_.szSession;
	char msg[256] = { 0 };
	sprintf_s(msg, 256, "PostTask: %s\n", str_session.c_str());
	OutputDebugStringA(msg);
	if (str_session == "") {
		return result;
	}
	EnterCriticalSection(&m_csForClientInstList);
	do {
		if (m_clientInstList.empty()) {
			break;
		}
		std::map<std::string, cmtd::CLIENT_CONTEXT_PTR>::iterator iter;
		iter = m_clientInstList.find(str_session);
		if (iter != m_clientInstList.end()) {
			cmtd::CLIENT_CONTEXT_PTR pClientContext = iter->second;
			if (pClientContext && pClientContext->clt_sock_num != INVALID_SOCKET) {
				cmtd::MEMOBUS_HEAD pshHead;
				int nPshHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
				int nPshParamLen = sizeof(cmtd::POST_PARAMETER);
				MAKE_HEAD(pshHead);
				pshHead.uiCmdType = cmtd::CMD_POST_PSH;
				pshHead.ulCmdLen = postParam_.nDataLen + nPshParamLen;
				pshHead.ulSerialNo = m_nSerialNo++;
				pshHead.ulTimeStamp = static_cast<unsigned long>(time(NULL));
				int nBufLen = nPshHeadLen + pshHead.ulCmdLen + 1;
				unsigned char * pSendBuf = new unsigned char[nBufLen];
				memset(pSendBuf, 0, nBufLen);
				int offset = 0;
				memcpy_s(pSendBuf, nBufLen, &pshHead, nPshHeadLen);
				offset += nPshHeadLen;
				memcpy_s(pSendBuf + offset, nBufLen - offset, &postParam_, nPshParamLen);
				offset += nPshParamLen;
				memcpy_s(pSendBuf + offset, nBufLen - offset, postParam_.pData, postParam_.nDataLen);
				offset += postParam_.nDataLen;
				if (send(pClientContext->clt_sock_num, reinterpret_cast<const char *>(pSendBuf),
					offset, 0) != SOCKET_ERROR) {
					result = 0;
				}
				RELEASE_ARRAY(pSendBuf);
			}
		}
	} while (0);
	LeaveCriticalSection(&m_csForClientInstList);
	return result;
}

void cmtd::TcpClient::SetMessageCallback(cmtd::MESSAGE_CALLBACK_PARAMETER msgCBParam_)
{
	std::string str_session = msgCBParam_.szSession;
	EnterCriticalSection(&m_csForClientInstList);
	do {
		if (str_session == "") {
			break;
		}
		std::map<std::string, cmtd::CLIENT_CONTEXT_PTR>::iterator iter;
		iter = m_clientInstList.find(str_session);
		if (iter != m_clientInstList.end()) {
			cmtd::CLIENT_CONTEXT_PTR pClientContext = iter->second;
			pClientContext->msgCB = msgCBParam_.msgCb;
			pClientContext->userDataForMsgCB = msgCBParam_.userData;
		}
	} while (0);
	LeaveCriticalSection(&m_csForClientInstList);
}

void cmtd::TcpClient::add_client_inst(std::string session_, cmtd::CLIENT_CONTEXT_PTR pClientContext_)
{
	if (session_ != "") {
		EnterCriticalSection(&m_csForClientInstList);
		if (m_clientInstList.empty()) {
			m_clientInstList.insert(std::make_pair(session_, pClientContext_));
			WakeConditionVariable(&m_cvForListNotEmpty);
		}
		else {
			std::map<std::string, cmtd::CLIENT_CONTEXT_PTR>::iterator iter;
			iter = m_clientInstList.find(session_);
			if (iter != m_clientInstList.end()) {
				cmtd::CLIENT_CONTEXT_PTR pContext = iter->second;
				pContext->clt_register_time = pClientContext_->clt_register_time;
			}
			else {
				m_clientInstList.insert(std::make_pair(session_, pClientContext_));
			}
		}
		LeaveCriticalSection(&m_csForClientInstList);
	}
}

void cmtd::TcpClient::remove_client_inst(std::string session_)
{
	if (session_ != "") {
		EnterCriticalSection(&m_csForClientInstList);
		if (!m_clientInstList.empty()) {
			std::map<std::string, cmtd::CLIENT_CONTEXT_PTR>::iterator iter;
			iter = m_clientInstList.find(session_);
			if (iter != m_clientInstList.end()) {
				cmtd::CLIENT_CONTEXT_PTR pContext = iter->second;
				RELEASE(pContext);
				iter = m_clientInstList.erase(iter);
			}
		}
		LeaveCriticalSection(&m_csForClientInstList);
	}
}

void cmtd::TcpClient::clear_client_insts()
{
	EnterCriticalSection(&m_csForClientInstList);
	if (!m_clientInstList.empty()) {
		std::map<std::string, cmtd::CLIENT_CONTEXT_PTR>::iterator iter;
		for (iter = m_clientInstList.begin(); iter != m_clientInstList.end(); iter++) {
			cmtd::CLIENT_CONTEXT_PTR pContext = iter->second;
			RELEASE(pContext);
		}
		m_clientInstList.clear();
	}
	LeaveCriticalSection(&m_csForClientInstList);
}

void cmtd::TcpClient::doWorkerJob()
{
	OVERLAPPED * pOverlapped;
	DWORD dwBytesTransfered = 0;
	DWORD dwFlags = 0;
	cmtd::CLIENT_CONTEXT_PTR pClientContext = NULL;
	BOOL bRet;
	do {
		DWORD dwRetEvent = WaitForSingleObject(m_hExitEvent, 100);
		if (dwRetEvent == WAIT_OBJECT_0) {
			break;
		}
		bRet = GetQueuedCompletionStatus(m_iocp, &dwBytesTransfered, (PULONG_PTR)&pClientContext,
			&pOverlapped, INFINITE);
		if (pClientContext == NULL) {
			break;
		}
		if (!bRet) {

			continue;
		}
		IO_CONTEXT_PTR pIoContext = CONTAINING_RECORD(pOverlapped, IO_CONTEXT, overlapped);
		if ((dwBytesTransfered == 0) && (pIoContext->io_type == IO_RECV)) {
			continue;
		}
		else {
			if (pIoContext->io_type == IO_RECV) {
				EnterCriticalSection(&m_csForClientInstList);
				if (!pClientContext->need_continue) {
					cmtd::MEMOBUS_HEAD head;
					int nHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
					int offset = 0;
					memcpy_s(&head, nHeadLen, pIoContext->wsaBuf.buf, nHeadLen);
					offset += nHeadLen;
					pClientContext->buffer_size = head.ulCmdLen + nHeadLen;
					pClientContext->buffer = new unsigned char[pClientContext->buffer_size];
					memset(pClientContext->buffer, 0, pClientContext->buffer_size);
					if (pClientContext->buffer_size <= (int)pIoContext->wsaBuf.len) {
						pClientContext->need_continue = false;
						memcpy_s(pClientContext->buffer, pClientContext->buffer_size, pIoContext->wsaBuf.buf,
							pClientContext->buffer_size);
						pClientContext->buffer_offset = pClientContext->buffer_size;
						cmtd::PARSE_INFO_PTR pParseInfo = new PARSE_INFO();
						pParseInfo->ulDataBufLen = pClientContext->buffer_size;
						pParseInfo->pDataBuf = new unsigned char[pParseInfo->ulDataBufLen];
						memset(pParseInfo->pDataBuf, 0, pParseInfo->ulDataBufLen);
						memcpy_s(pParseInfo->pDataBuf, pParseInfo->ulDataBufLen, pClientContext->buffer, 
							pClientContext->buffer_size);
						pParseInfo->pTcpClient = this;
						pParseInfo->pClientContext = pClientContext;
						pParseInfo->pIoContext = pIoContext;
						threadpool_add_job(m_pool, parse_thread, pParseInfo);
						Sleep(5);
						pClientContext->reset_buffer();
					}
					else {
						pClientContext->need_continue = true;
						memcpy_s(pClientContext->buffer, pIoContext->wsaBuf.len, pIoContext->wsaBuf.buf,
							pIoContext->wsaBuf.len);
						pClientContext->buffer_offset += pIoContext->wsaBuf.len;
					}
				}
				else {
					if (pClientContext->buffer_size - pClientContext->buffer_offset <= (int)pIoContext->wsaBuf.len) {
						pClientContext->need_continue = false;
						memcpy_s(pClientContext->buffer + pClientContext->buffer_offset,
							pClientContext->buffer_size - pClientContext->buffer_offset,
							pIoContext->wsaBuf.buf, pClientContext->buffer_size - pClientContext->buffer_offset);
						pClientContext->buffer_offset = pClientContext->buffer_size;
						cmtd::PARSE_INFO_PTR pParseInfo = new cmtd::PARSE_INFO();
						pParseInfo->pTcpClient = this;
						pParseInfo->pClientContext = pClientContext;
						pParseInfo->pIoContext = pIoContext;
						pParseInfo->ulDataBufLen = pClientContext->buffer_size;
						pParseInfo->pDataBuf = new unsigned char[pParseInfo->ulDataBufLen];
						memset(pParseInfo->pDataBuf, 0, pParseInfo->ulDataBufLen);
						memcpy_s(pParseInfo->pDataBuf, pParseInfo->ulDataBufLen, pClientContext->buffer,
							pClientContext->buffer_size);
						threadpool_add_job(m_pool, parse_thread, pParseInfo);
						Sleep(5);
						pClientContext->reset_buffer();
					}
					else {
						pClientContext->need_continue = true;
						memcpy_s(pClientContext->buffer + pClientContext->buffer_offset,
							pIoContext->wsaBuf.len, pIoContext->wsaBuf.buf, pIoContext->wsaBuf.len);
						pClientContext->buffer_offset += pIoContext->wsaBuf.len;
					}
				}
				LeaveCriticalSection(&m_csForClientInstList);			

				memset(&pIoContext->overlapped, 0, sizeof(OVERLAPPED));
				pIoContext->reset_buffer();
				pIoContext->io_type = IO_RECV;
				WSARecv(pClientContext->clt_sock_num, &pIoContext->wsaBuf, 1, &dwBytesTransfered,
					&dwFlags, &pIoContext->overlapped, NULL);
			}
		}
	} while (1);
}

void cmtd::TcpClient::doKeepAliveJob()
{
	do {
		DWORD dwRet = WaitForSingleObject(m_hExitEvent, 100);
		if (dwRet == WAIT_OBJECT_0) {
			break;
		}
		EnterCriticalSection(&m_csForClientInstList);
		if (m_clientInstList.empty() && !m_bQuit) {
			SleepConditionVariableCS(&m_cvForListNotEmpty, &m_csForClientInstList, INFINITE);
		}
		if (m_bQuit) {
			LeaveCriticalSection(&m_csForClientInstList);
			break;
		}
		std::map<std::string, cmtd::CLIENT_CONTEXT_PTR>::iterator iter;
		for (iter = m_clientInstList.begin(); iter != m_clientInstList.end(); iter++) {
			cmtd::CLIENT_CONTEXT_PTR pContext = iter->second;
			std::string str_session = iter->first;
			if (pContext && pContext->clt_sock_num != INVALID_SOCKET) {
				time_t now = time(NULL);
				time_t last_active_time = pContext->clt_last_active_time;
				if (abs(difftime(now, last_active_time)) > 60.0) {
					cmtd::MEMOBUS_HEAD aliveHead;
					unsigned long ulHeadLen = sizeof(aliveHead);
					MAKE_HEAD(aliveHead);
					aliveHead.uiVersion = INTERNAL_VERSION;
					aliveHead.uiCmdType = cmtd::CMD_KEEPALIVE;
					aliveHead.ulSerialNo = m_nSerialNo++;
					aliveHead.ulTimeStamp = (unsigned long)now;
					cmtd::KEEPALIVE_PARAMETER aliveParam;
					unsigned long ulParamLen = sizeof(aliveParam);
					aliveHead.ulCmdLen = ulParamLen;
					strncpy_s(aliveParam.szSession, sizeof(aliveParam.szSession),
						str_session.c_str(), str_session.size());
					unsigned long ulBufLen = ulHeadLen + ulParamLen + 1;
					unsigned char * pSendBuf = new unsigned char[ulBufLen];
					memset(pSendBuf, 0, ulBufLen);
					int offset = 0;
					memcpy_s(pSendBuf, ulHeadLen, &aliveHead, ulHeadLen);
					offset += ulHeadLen;
					memcpy_s(pSendBuf + offset, ulParamLen, &aliveParam, ulParamLen);
					offset += ulParamLen;
					if (send(pContext->clt_sock_num, (const char *)pSendBuf, offset, 0) == SOCKET_ERROR) {
						int sockErr = WSAGetLastError();
					}
					pContext->clt_last_active_time = (long)now;
					RELEASE_ARRAY(pSendBuf);
				}
			}
		}
		LeaveCriticalSection(&m_csForClientInstList);
		Sleep(50);
	} while (1);
}

bool cmtd::TcpClient::associationWithCompletionPort(SOCKET sock_, HANDLE hCompletionPort_, 
	DWORD dwCompletionKey_)
{
	HANDLE h = CreateIoCompletionPort((HANDLE)sock_, hCompletionPort_, dwCompletionKey_, 0);
	return (h == hCompletionPort_);
}

void cmtd::TcpClient::doParseJob(unsigned char * pDataBuf, unsigned long ulDataBufLen, 
	cmtd::CLIENT_CONTEXT_PTR pClientContext_, cmtd::IO_CONTEXT_PTR pIoContext_)
{
	do {
		if (pDataBuf != NULL && ulDataBufLen > 0) {
			cmtd::MEMOBUS_HEAD respHead;
			int offset = 0;
			int nRespHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
			memcpy_s(&respHead, nRespHeadLen, pDataBuf, nRespHeadLen);
			offset += nRespHeadLen;
			cmtd::ResponseInfo respInfo;
			respInfo.pRespData = pDataBuf;
			respInfo.ulDataLen = ulDataBufLen;
			if (check_valid_head(respHead)) {
				switch (respHead.uiCmdType) {
				case cmtd::CMD_REGISTER_RESP:
					{
						deal_register_response(respInfo, pClientContext_, pIoContext_);
					}
					break;
				case cmtd::CMD_UNREGISTER_RESP:
					{
						deal_unregister_response(respInfo, pClientContext_, pIoContext_);
					}
					break;
				case cmtd::CMD_FETCH_RESP:
					{
						deal_fetch_response(respInfo, pClientContext_, pIoContext_);
					}
					break;
				case cmtd::CMD_FINISH_RESP:
					{
						deal_finish_response(respInfo, pClientContext_, pIoContext_);
					}
					break;
				case cmtd::CMD_KEEPALIVE:
					{
						deal_alive_response(respInfo, pClientContext_, pIoContext_);
					}
					break;
				case cmtd::CMD_NOTIFY_PSH:
					{
						deal_notify_response(respInfo, pClientContext_, pIoContext_);
					}
					break;
				case cmtd::CMD_POST_PSH:
					{
						deal_post_push(respInfo, pClientContext_, pIoContext_);
					}
					break;
				default:
					break;
				}
			}
		}
	} while (0);
}

void cmtd::TcpClient::deal_register_response(cmtd::ResponseInfo respInfo_, 
	cmtd::CLIENT_CONTEXT_PTR pClientContext_, cmtd::IO_CONTEXT_PTR pIoContext_)
{
	do {
		if (!respInfo_.pRespData || respInfo_.ulDataLen == 0) {
			break;
		}
		cmtd::MEMOBUS_HEAD respHead;
		unsigned long nRespHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
		int offset = 0;
		memcpy_s(&respHead, nRespHeadLen, respInfo_.pRespData, nRespHeadLen);
		offset += nRespHeadLen;
		cmtd::REGISTER_RESPONSE_PARAMETER regRespParam;
		unsigned long nRegRespParamLen = sizeof(cmtd::REGISTER_RESPONSE_PARAMETER);
		if (nRegRespParamLen == respHead.ulCmdLen 
			&& respInfo_.ulDataLen >= nRegRespParamLen + nRespHeadLen) {
			memcpy_s(&regRespParam, nRegRespParamLen, respInfo_.pRespData + offset, nRegRespParamLen);
			offset += nRegRespParamLen;
			cmtd::MESSAGE_INFO msgInfo;
			msgInfo.nMsgType = cmtd::CMD_REGISTER_RESP;
			msgInfo.nResult = regRespParam.nResult;
			if (regRespParam.nResult == 0 && regRespParam.register_parameter.szSession != NULL) {	
				std::string str_session = regRespParam.register_parameter.szSession;
				if (pClientContext_->clt_session == "") {
					pClientContext_->clt_session = str_session;
				}
				if (str_session == pClientContext_->clt_session) {
					msgInfo.userData = pClientContext_->userDataForMsgCB;
					strncpy_s(msgInfo.szSession, sizeof(msgInfo.szSession), str_session.c_str(),
						str_session.size());
					if (pClientContext_->msgCB) {
						pClientContext_->msgCB(msgInfo);
					}
				}
			}
		}
	} while (0);
}

void cmtd::TcpClient::deal_alive_response(cmtd::ResponseInfo respInfo_, 
	cmtd::CLIENT_CONTEXT_PTR pClientContext_, cmtd::IO_CONTEXT_PTR pIoContext_)
{
	do {
		cmtd::MEMOBUS_HEAD respHead;
		unsigned long nRespHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
		int offset = 0;
		if (respInfo_.pRespData != NULL && respInfo_.ulDataLen > nRespHeadLen) {
			memcpy_s(&respHead, nRespHeadLen, respInfo_.pRespData, nRespHeadLen);
			cmtd::KEEPALIVE_PARAMETER aliveParam;
			unsigned long nAliveParamLen = sizeof(cmtd::KEEPALIVE_PARAMETER);
			offset += nRespHeadLen;
			if (respHead.ulCmdLen == nAliveParamLen && respInfo_.ulDataLen >= nAliveParamLen + nRespHeadLen) {
				memcpy_s(&aliveParam, nAliveParamLen, respInfo_.pRespData + offset, nAliveParamLen);
				std::string str_session = aliveParam.szSession;
				EnterCriticalSection(&m_csForClientInstList);
				if (str_session == pClientContext_->clt_session) {
					pClientContext_->clt_last_active_time = static_cast<long>(time(NULL));
				}
				LeaveCriticalSection(&m_csForClientInstList);
			}
		}
	} while (0);
}

void cmtd::TcpClient::deal_fetch_response(cmtd::ResponseInfo respInfo_, 
	cmtd::CLIENT_CONTEXT_PTR pClientContext_, cmtd::IO_CONTEXT_PTR pIoContext_)
{
	cmtd::MEMOBUS_HEAD respHead;
	unsigned long nRespHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
	int offset = 0;
	char msg[256] = { 0 };
	sprintf_s(msg, 256, "%s: data length=%d\n", __FUNCTION__, respInfo_.ulDataLen);
	OutputDebugStringA(msg);
	do {
		if (respInfo_.pRespData != NULL && respInfo_.ulDataLen > nRespHeadLen) {
			memcpy_s(&respHead, nRespHeadLen, respInfo_.pRespData, nRespHeadLen);
			offset += nRespHeadLen;
			cmtd::FETCH_RESPONSE_PARAMETER fetchRespParam;
			unsigned long nFetchRespParamLen = sizeof(cmtd::FETCH_RESPONSE_PARAMETER);
			memcpy_s(&fetchRespParam, nFetchRespParamLen, respInfo_.pRespData + offset, 
				nFetchRespParamLen);
			offset += nFetchRespParamLen;
			cmtd::MESSAGE_INFO msgInfo;
			msgInfo.nMsgType = cmtd::CMD_FETCH_RESP;
			msgInfo.nResult = fetchRespParam.nResult;
			strncpy_s(msgInfo.szSession, sizeof(msgInfo.szSession), fetchRespParam.szSession,
				strlen(fetchRespParam.szSession));
			EnterCriticalSection(&m_csForClientInstList);
			if (fetchRespParam.nResult == 0) {
				unsigned long ulFetchDataLen = respInfo_.ulDataLen - nRespHeadLen;
				unsigned char * pFetchData = new unsigned char[ulFetchDataLen];
				memcpy_s(pFetchData, ulFetchDataLen, respInfo_.pRespData + nRespHeadLen, ulFetchDataLen);
				std::string str_session = fetchRespParam.szSession;			
				if (str_session == pClientContext_->clt_session) {
					msgInfo.userData = pClientContext_->userDataForMsgCB;
					if (pClientContext_->msgCB) {
						msgInfo.msgDataLen = respInfo_.ulDataLen;
						msgInfo.msgData = pFetchData;
						pClientContext_->msgCB(msgInfo);
					}
				}	
				RELEASE_ARRAY(pFetchData);
			}
			else {
				if (pClientContext_->msgCB) {
					pClientContext_->msgCB(msgInfo);
				}
			}
			LeaveCriticalSection(&m_csForClientInstList);
		}		
	} while (0);
}

void cmtd::TcpClient::deal_finish_response(cmtd::ResponseInfo respInfo_, 
	cmtd::CLIENT_CONTEXT_PTR pClientContext_, cmtd::IO_CONTEXT_PTR pIoContext_)
{
	cmtd::MEMOBUS_HEAD respHead;
	unsigned long nRespHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
	int offset = 0;
	do {
		if (respInfo_.pRespData != NULL && respInfo_.ulDataLen > nRespHeadLen) {
			memcpy_s(&respHead, nRespHeadLen, respInfo_.pRespData, nRespHeadLen);
			offset += nRespHeadLen;
			cmtd::FINISH_RESPONSE_PARAMETER finishRespParam;
			unsigned long nFinishRespParamLen = sizeof(cmtd::FINISH_RESPONSE_PARAMETER);
			if (respHead.ulCmdLen == nFinishRespParamLen &&
				respInfo_.ulDataLen >= nFinishRespParamLen + nRespHeadLen) {
				memcpy_s(&finishRespParam, nFinishRespParamLen, respInfo_.pRespData + offset,
					nFinishRespParamLen);
				offset += nFinishRespParamLen;
				cmtd::MESSAGE_INFO msgInfo;
				msgInfo.nMsgType = cmtd::CMD_FINISH_RESP;
				msgInfo.nResult = finishRespParam.nResult;
				strncpy_s(msgInfo.szSession, sizeof(msgInfo.szSession), finishRespParam.szSession,
					strlen(finishRespParam.szSession));
				std::string str_session = finishRespParam.szSession;
				EnterCriticalSection(&m_csForClientInstList);
				if (str_session == pClientContext_->clt_session) {
					msgInfo.userData = pClientContext_->userDataForMsgCB;
					if (pClientContext_->msgCB) {
						pClientContext_->msgCB(msgInfo);
					}
				}
				LeaveCriticalSection(&m_csForClientInstList);
			}
		}
	} while (0);
}

void cmtd::TcpClient::deal_unregister_response(cmtd::ResponseInfo respInfo_, 
	cmtd::CLIENT_CONTEXT_PTR pClientContext_, cmtd::IO_CONTEXT_PTR pIoContext_)
{
	cmtd::MEMOBUS_HEAD respHead;
	unsigned long nRespHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
	int offset = 0;
	do {
		if (respInfo_.pRespData != NULL && respInfo_.ulDataLen > nRespHeadLen) {
			memcpy_s(&respHead, nRespHeadLen, respInfo_.pRespData, nRespHeadLen);
			offset += nRespHeadLen;
			cmtd::UNREGISTER_RESPONSE_PARAMETER unRegRespParam;
			unsigned long nUnRegRespParamLen = sizeof(cmtd::UNREGISTER_RESPONSE_PARAMETER);
			if (nUnRegRespParamLen == respHead.ulCmdLen &&
				respInfo_.ulDataLen >= nUnRegRespParamLen + nRespHeadLen) {
				memcpy_s(&unRegRespParam, nUnRegRespParamLen, respInfo_.pRespData + offset,
					nUnRegRespParamLen);
				//unRegRespParam.nResult
				cmtd::MESSAGE_INFO msgInfo;
				msgInfo.nMsgType = cmtd::CMD_UNREGISTER_RESP;
				msgInfo.nResult = unRegRespParam.nResult;
				strncpy_s(msgInfo.szSession, sizeof(msgInfo.szSession),
					unRegRespParam.unregister_parameter.szSession,
					strlen(unRegRespParam.unregister_parameter.szSession));
				std::string str_session = unRegRespParam.unregister_parameter.szSession;
				EnterCriticalSection(&m_csForClientInstList);
				if (str_session == pClientContext_->clt_session) {
					msgInfo.userData = pClientContext_->userDataForMsgCB;
					if (pClientContext_->msgCB) {
						pClientContext_->msgCB(msgInfo);
					}
				}
				if (m_clientInstList.empty()) {
					std::map<std::string, cmtd::CLIENT_CONTEXT_PTR>::iterator iter;
					iter = m_clientInstList.find(str_session);
					if (iter != m_clientInstList.end()) {
						remove_client_inst(str_session);
						RELEASE(pClientContext_);
						m_clientInstList.erase(iter);
					}
				}
				LeaveCriticalSection(&m_csForClientInstList);
			}
		}
	} while (0);
}

void cmtd::TcpClient::deal_notify_response(cmtd::ResponseInfo respInfo_, 
	cmtd::CLIENT_CONTEXT_PTR pClientContext_, cmtd::IO_CONTEXT_PTR pIoContext_)
{
	cmtd::MEMOBUS_HEAD resqHead;
	unsigned long nRespHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
	int offset = 0;
	do {
		if (respInfo_.pRespData != NULL && respInfo_.ulDataLen > nRespHeadLen) {
			memcpy_s(&resqHead, nRespHeadLen, respInfo_.pRespData, nRespHeadLen);
			cmtd::NOTIFY_PARAMETER notifyParam;
			unsigned long nNotifyParamLen = sizeof(cmtd::NOTIFY_PARAMETER);
			offset += nRespHeadLen;
			if (resqHead.ulCmdLen == nNotifyParamLen && 
				respInfo_.ulDataLen >= nRespHeadLen + nNotifyParamLen) {
				memcpy_s(&notifyParam, nNotifyParamLen, respInfo_.pRespData + offset, nNotifyParamLen);
				offset += nNotifyParamLen;
				std::string str_session = notifyParam.szSession;
				if (str_session == pClientContext_->clt_session) {
					cmtd::MESSAGE_INFO msgInfo;
					msgInfo.nMsgType = cmtd::CMD_NOTIFY_PSH;
					strncpy_s(msgInfo.szSession, sizeof(msgInfo.szSession), str_session.c_str(),
						str_session.size());
					if (pClientContext_->msgCB) {
						msgInfo.userData = pClientContext_->userDataForMsgCB;
						pClientContext_->msgCB(msgInfo);
					}
				}
			}
		}
	} while (0);
}

void cmtd::TcpClient::deal_post_push(cmtd::ResponseInfo respInfo_, 
	cmtd::CLIENT_CONTEXT_PTR pClientContext_, cmtd::IO_CONTEXT_PTR pIoContext_)
{
	cmtd::MEMOBUS_HEAD pshHead;
	unsigned long nPshHeadLen = sizeof(cmtd::MEMOBUS_HEAD);
	int offset = 0;
	do {
		if (respInfo_.pRespData != NULL && respInfo_.ulDataLen > nPshHeadLen) {
			memcpy_s(&pshHead, nPshHeadLen, respInfo_.pRespData, nPshHeadLen);
			offset += nPshHeadLen;
			if (respInfo_.ulDataLen >= pshHead.ulCmdLen + nPshHeadLen) {
				cmtd::POST_PARAMETER postParam;
				unsigned long nPostParamLen = sizeof(cmtd::POST_PARAMETER);
				memcpy_s(&postParam, nPostParamLen, respInfo_.pRespData + offset,
					nPostParamLen);
				offset += nPostParamLen;
				std::string str_session = postParam.szSession;
				if (postParam.nDataLen > 0) {
					unsigned char * pPostData = new unsigned char[postParam.nDataLen];
					memset(pPostData, 0, nPostParamLen + 1);
					memcpy_s(pPostData, postParam.nDataLen, respInfo_.pRespData + offset,
						postParam.nDataLen);
					cmtd::MESSAGE_INFO msgInfo;
					msgInfo.nMsgType = cmtd::CMD_POST_PSH;
					strncpy_s(msgInfo.szSession, sizeof(msgInfo.szSession), str_session.c_str(),
						str_session.size());
					msgInfo.msgDataLen = postParam.nDataLen;
					msgInfo.msgData = pPostData;
					if (str_session == pClientContext_->clt_session) {
						if (pClientContext_->msgCB) {
							msgInfo.userData = pClientContext_->userDataForMsgCB;
							pClientContext_->msgCB(msgInfo);
						}
					}
					RELEASE_ARRAY(pPostData);
				}
			}
		}
	} while (0);
}

bool cmtd::TcpClient::check_valid_head(cmtd::MEMOBUS_HEAD head_)
{
	if ((head_.ucMark[0] == 'C') && (head_.ucMark[1] == 'M') && (head_.ucMark[2] = 'T')
		&& head_.ucMark[3] == 'D' && head_.uiVersion == INTERNAL_VERSION) {
		return true;
	}
	return false;
}

unsigned int __stdcall cmtd::keep_alive_thread(void * param)
{
	cmtd::TcpClient * pInst = (cmtd::TcpClient *)param;
	if (pInst) {
		pInst->doKeepAliveJob();
	}
	return 0;
}

unsigned int __stdcall cmtd::worker_thread(void * param)
{
	cmtd::TcpClient * pInst = (cmtd::TcpClient *)param;
	if (pInst) {
		pInst->doWorkerJob();
	}
	return 0;
}

unsigned int __stdcall cmtd::parse_thread(void * param)
{
	PARSE_INFO_PTR pParseInfo = (PARSE_INFO_PTR)param;
	if (pParseInfo) {
		cmtd::TcpClient * pTcInst = pParseInfo->pTcpClient;
		pTcInst->doParseJob(pParseInfo->pDataBuf, pParseInfo->ulDataBufLen, 
			pParseInfo->pClientContext, pParseInfo->pIoContext);
		Sleep(10);
		RELEASE(pParseInfo);
	}
	return 0;
}